//
// Created by 17775 on 2024/8/2.
//

#ifndef RAYCI_DATAPROCESSOR_H
#define RAYCI_DATAPROCESSOR_H


#include <functional>
#include <memory>
#include "interruptible_thread.h"
#include "threadsafe_linkedqueue.h"
template<class ResultType>
struct TaskItem final {
    std::function<ResultType()> _task;
    std::function<void(const ResultType&)> _onCompleted;
    TaskItem() = default;
    template<class FunctionType1, class FunctionType2,
            typename = std::enable_if_t<std::is_same_v < ResultType, std::remove_reference_t<decltype(std::declval<FunctionType1>()())>> || (std::is_fundamental_v<ResultType> && std::is_fundamental_v <std::remove_reference_t<decltype(std::declval<FunctionType1>()())>>)>>
    TaskItem(FunctionType1&& task, FunctionType2&& onCompleted) :
            _task(std::forward<FunctionType1>(task)), _onCompleted(std::forward<FunctionType2>(onCompleted)) {
    }
    ResultType operator()() {
        try {
            ResultType r =  _task();
            _onCompleted(r);
            return r;
        }
        catch (...) {
            return ResultType();
        }
    }
};
template<class ResultType>
class DataProcessor {
protected:
    threadsafe_linkedqueue<TaskItem<ResultType>, std::condition_variable_any> queue;
    interruptible_thread _taskThread;
public:
    DataProcessor() : _taskThread([=]() {
        try {
            for ( ;; ) {
                TaskItem<ResultType> taskItem;
                if(this->queue.wait_and_pop(taskItem)) {
                    ResultType result = taskItem();
                }
                else {
                }
            }
        }
        catch (...) {
        }
    }){
    }
    ~DataProcessor()  {
        if(_taskThread.joinable()) {
            _taskThread.join();
        }
    }
    template<class FunctionType1, class FunctionType2, typename R = std::remove_reference_t<decltype(std::declval<FunctionType1>()())>,
            typename = decltype(std::declval<FunctionType2>()(std::declval<FunctionType1>()()))>
    void submit(FunctionType1&& task, FunctionType2&& onCompleted)  {
        queue.push(TaskItem<ResultType>(std::forward<FunctionType1>(task), std::forward<FunctionType2>(onCompleted)));
    }
    void interrupt() {
        _taskThread.interrupt();
    }
};

//#include <functional>
//#include <memory>
//#include "interruptible_thread.h"
//#include "threadsafe_linkedqueue.h"
//struct TaskItem final {
//    struct impl_base {
//        virtual void call()=0;
//        virtual ~impl_base() = default;
//    };
//    template<class FunctionType>
//    struct impl_type: impl_base
//    {
//        FunctionType _task;
//        explicit impl_type(FunctionType&& task): _task(std::forward<FunctionType>(task)) {
//        }
//        void call() override {
//            try {
//                _task();
//            }
//            catch (...) {
//            }
//        }
//    };
//    std::unique_ptr<impl_base> impl;
//    TaskItem() = default;
//    template<class FunctionType>
//    explicit TaskItem(FunctionType&& task) :
//            impl(std::make_unique<impl_type<FunctionType>>(std::forward<FunctionType>(task))) {
//    }
//    void operator()() const {
//        impl->call();
//    }
//    TaskItem(TaskItem&& other) noexcept :
//            impl(std::move(other.impl))
//    {}
//    TaskItem& operator=(TaskItem&& other) noexcept {
//        impl=std::move(other.impl);
//        return *this;
//    }
//    TaskItem(const TaskItem&)=delete;
//    TaskItem(TaskItem&)=delete;
//    TaskItem& operator=(const TaskItem&)=delete;
//};
//class DataProcessor {
//protected:
//    threadsafe_linkedqueue<TaskItem, std::condition_variable_any> pool_work_queue;
//    std::vector<interruptible_thread> threads;
//    join_interruptible_threads joiner;
//    void worker_thread() {
//        try {
//            for ( ;; ) {
//                TaskItem taskItem;
//                if(this->pool_work_queue.wait_and_pop(taskItem)) {
//                    taskItem();
//                }
//                else {
//                }
//            }
//        }
//        catch (...) {
//        }
//    }
//private:
//    DataProcessor() :joiner(threads) {
//        unsigned const thread_count = std::thread::hardware_concurrency();
//        threads.reserve(thread_count);
//        try {
//            for(unsigned i = 0; i < thread_count; ++i) {
//                threads.emplace_back([=] {
//                    this->worker_thread();
//                });
//            }
//        }
//        catch(...) {
//            throw;
//        }
//    }
//public:
//    ~DataProcessor()  = default;
//    template<class FunctionType>
//    std::future<typename std::result_of<FunctionType()>::type> submit(FunctionType&& f)  {
//        typedef typename std::result_of<FunctionType()>::type result_type;
//        std::packaged_task<result_type()> task(std::forward<FunctionType>(f));
//        std::future<result_type> res(task.get_future());
//        pool_work_queue.push(TaskItem(std::move(task)));
//        return res;
//    }
//    static DataProcessor& getDataProcessor() {
//        static DataProcessor dataProcessor;
//        return dataProcessor;
//    }
//};
//
//class join_interruptible_threads
//{
//    std::vector<interruptible_thread>& threads;
//public:
//    explicit join_interruptible_threads(std::vector<interruptible_thread>& threads_):
//            threads(threads_)
//    {}
//    ~join_interruptible_threads()
//    {
//        auto s = threads.size();
//        for(unsigned long i=0; i<s; ++i)
//        {
//            threads[i].interrupt();
//        }
//        for(unsigned long i=0; i<s; ++i)
//        {
//            if(threads[i].joinable())
//                threads[i].join();
//        }
//    }
//};


#endif //RAYCI_DATAPROCESSOR_H
